package com.shujia.flink

import org.apache.flink.api.common.typeinfo.{TypeInformation, Types}
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.scala.StreamTableEnvironment
import org.apache.flink.table.api.scala._
import org.apache.flink.api.scala._
import org.apache.flink.core.fs.FileSystem.WriteMode
import org.apache.flink.table.sinks.CsvTableSink
import org.apache.flink.types.Row

object Demo7StreamTableApi {
  def main(args: Array[String]): Unit = {
    val streamEnv = StreamExecutionEnvironment.getExecutionEnvironment

    val tableEnv = StreamTableEnvironment.create(streamEnv)

    val ds = streamEnv

      /**
        * a,1
        * a,2
        */
      .socketTextStream("node1", 7777)
      .map(_.split(","))
      .map(split => Event(split(0), split(1).toInt))


    val table = tableEnv.fromDataStream(ds)


    //打印结果
    table.select('word)
      //打印输出
      .toAppendStream[Row] //.print()

    /**
      * toAppendStream  只适合追加的l流
      * toRetractStream  可以用于任何流
      *
      */

    table
      .groupBy('word)
      .select('word, 'num.sum)
      .toRetractStream[Row] //.print()


    //将实时流注册成一张表
    tableEnv.registerTable("words", table)

    val resultTable = tableEnv.sqlQuery(
      """
        |select word,sum(num) as s from words group by word
        |
      """.stripMargin)

    resultTable.toRetractStream[Row] //.print()

    val wDS = streamEnv.fromCollection(List(
      W("a", 1),
      W("b", 1),
      W("c", 1)
    ))
    //将ds注册成一张表
    tableEnv.registerDataStream("w", wDS)

    tableEnv.sqlQuery(
      """
        |select * from words as a right join w as b on a.word=b.word
        |
      """.stripMargin).toRetractStream[Row].print()


    //构建一个输出表
    val sink = new CsvTableSink("data/flink/words", ",", 1, WriteMode.OVERWRITE)
    val fieldNames: Array[String] = Array("word", "num")
    val fieldTypes: Array[TypeInformation[_]] = Array(Types.STRING, Types.INT)
    tableEnv.registerTableSink("ws", fieldNames, fieldTypes, sink)


    //实时将数据插入到文件
    table.insertInto("ws")

    ///  更新流不能写入到文件
    //    table
    //      .groupBy('word)
    //      .select('word, 'num.sum.as('num))
    //      .insertInto("ws")

    streamEnv.execute()

  }

  case class Event(word: String, num: Int)

  case class W(word: String, age: Int)

}
